#include "RouteInfoSelector.h"

namespace zeroplus::mq::consumer
{
    RouteInfoSelector::RouteInfoSelector()
        : index_(0)
    {
    }
    RouteInfoSelector::~RouteInfoSelector()
    {
    }

    std::tuple<routeinfo::QueueData, routeinfo::BrokerData, bool> RouteInfoSelector::SelectRouteInfo(const routeinfo::TopicRouteInfo &topicRouteInfo,
                                                                                                     LoadBalanceStrategy strategy,
                                                                                                     std::string messageKey)
    {
        auto [queueData, hasQueueData] = SelectQueue(topicRouteInfo, strategy, messageKey);
        if (!hasQueueData)
        {
            return std::make_tuple(routeinfo::QueueData(), routeinfo::BrokerData(), false);
        }
        auto [brokerData, hasBrokerData] = SelectBroker(topicRouteInfo, queueData.broker_name());
        if (!hasBrokerData)
        {
            return std::make_tuple(routeinfo::QueueData(), routeinfo::BrokerData(), false);
        }
        return std::make_tuple(queueData, brokerData, true);
    }

    std::tuple<routeinfo::BrokerData, bool> RouteInfoSelector::SelectBroker(const routeinfo::TopicRouteInfo &topicRouteInfo, const std::string &brokerName)
    {
        for (int i = 0; i < topicRouteInfo.broker_data_list_size(); ++i)
        {
            const routeinfo::BrokerData &brokerData = topicRouteInfo.broker_data_list(i);
            if (brokerData.broker_name() == brokerName)
            {
                return std::make_tuple(brokerData, true);
            }
        }
        return std::make_tuple(routeinfo::BrokerData(), false);
    }

    std::tuple<routeinfo::QueueData, bool> RouteInfoSelector::SelectQueue(const routeinfo::TopicRouteInfo &topicRouteInfo, LoadBalanceStrategy strategy, std::string messageKey)
    {
        if (!topicRouteInfo.queue_data_list_size())
        {
            // LOG_ERROR("Queue data list is empty");
            return std::make_tuple(routeinfo::QueueData(), false);
        }

        switch (strategy)
        {
        case ROUND_ROBIN:
            return SelectQueueByRoundRobin(topicRouteInfo);
        case RANDOM:
            return SelectQueueByRandom(topicRouteInfo);
        case HASH:
            return SelectQueueByHash(topicRouteInfo, messageKey);
        default:
            LOG_FATAL("Unsupported load balance strategy: %d", strategy);
        }
    }

    std::tuple<routeinfo::QueueData, bool> RouteInfoSelector::SelectQueueByRoundRobin(const routeinfo::TopicRouteInfo &topicRouteInfo)
    {
        // LOG_DEBUG("topicRouteInfo.queue_data_list_size()=%d", topicRouteInfo.queue_data_list_size());
        int currentIndex = index_.fetch_add(1) % topicRouteInfo.queue_data_list_size();
        return std::make_tuple(topicRouteInfo.queue_data_list(currentIndex), true);
    }

    std::tuple<routeinfo::QueueData, bool> RouteInfoSelector::SelectQueueByRandom(const routeinfo::TopicRouteInfo &topicRouteInfo)
    {
        std::random_device rd;
        std::mt19937 gen(rd());
        std::uniform_int_distribution<> dis(0, topicRouteInfo.queue_data_list_size() - 1);
        int randomIndex = dis(gen);
        return std::make_tuple(topicRouteInfo.queue_data_list(randomIndex), true);
    }

    std::tuple<routeinfo::QueueData, bool> RouteInfoSelector::SelectQueueByHash(const routeinfo::TopicRouteInfo &topicRouteInfo, std::string &messageKey)
    {
        if (messageKey.empty())
        {
            LOG_ERROR("Message key is required for hash strategy");
            messageKey = "default";
        }
        // 计算哈希值
        std::hash<std::string> hasher;
        size_t hashCode = hasher(messageKey);
        int hashIndex = hashCode % topicRouteInfo.queue_data_list_size();
        return std::make_tuple(topicRouteInfo.queue_data_list(hashIndex), true);
    }
}